package org.example.com.atguigu.exercise;

import org.apache.commons.collections4.IterableUtils;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import scala.Tuple2;
import scala.Tuple3;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;

public class Exercise2{
    public static void main(String[] args) {
        // 统计出每个省份菜的种类数最多的三个农贸市场
        SparkConf conf = new SparkConf().setMaster("local[4]").setAppName("com.atguigu.day01.$01_RddCreate");
        JavaSparkContext sc = new JavaSparkContext(conf);
        // rdd1 [(西红柿	5.20	2018/1/1	四川南充市桑园坝农产品批发市场	四川	南充), ...]
        JavaRDD<String> rdd1 = sc.textFile("E:\\大数据第三阶段\\spark\\6.代码\\product.txt");
        // rdd2 去脏数据
        JavaRDD<String> rdd2 = rdd1.filter(new Function<String, Boolean>() {
            @Override
            public Boolean call(String v1) throws Exception {
                return v1.split("\t").length == 6;
            }
        });
        // rdd3 去重 [(山西,山西汾阳市晋阳农副产品批发市场,香菜), (山西,山西汾阳市晋阳农副产品批发市场,大葱), ...]
        JavaRDD<Tuple3<String, String, String>> rdd3 = rdd2.map(new Function<String, Tuple3<String, String, String>>() {
            @Override
            public Tuple3<String, String, String> call(String s) throws Exception {
                String k1 = s.split("\t")[4];
                String k2 = s.split("\t")[3];
                String v = s.split("\t")[0];
                return new Tuple3<>(k1, k2, v);
            }
        }).distinct();
        // rdd4 [((天津,天津金钟蔬菜批发市场),1), ((北京,北京昌平区水屯农副产品批发市场),1), ...]
        JavaPairRDD<Tuple2<String, String>, Integer> rdd4 = rdd3.mapToPair(new PairFunction<Tuple3<String, String, String>, Tuple2<String, String>, Integer>() {
            @Override
            public Tuple2<Tuple2<String, String>, Integer> call(Tuple3<String, String, String> v2) throws Exception {
                return new Tuple2<>(new Tuple2<>(v2._1(), v2._2()), 1);
            }
        });
        // rdd5 分组聚合
        JavaPairRDD<Tuple2<String, String>, Integer> rdd5 = rdd4.reduceByKey(new Function2<Integer, Integer, Integer>() {
            @Override
            public Integer call(Integer v1, Integer v2) throws Exception {
                return v1 + v2;
            }
        });
        // rdd6 按省份分组 [ (辽宁,[((辽宁,辽宁朝阳市果菜批发市场),37), ((辽宁,辽宁阜新市瑞轩蔬菜农副产品综合批发市场),53),...]), ...]
        JavaPairRDD<String, Iterable<Tuple2<Tuple2<String, String>, Integer>>> rdd6 = rdd5.groupBy(new Function<Tuple2<Tuple2<String, String>, Integer>, String>() {
            @Override
            public String call(Tuple2<Tuple2<String, String>, Integer> v1) throws Exception {
                return v1._1._1;
            }
        });
        // rdd7 按照农贸市场排序
        JavaPairRDD<String, List<String>> rdd7 = rdd6.mapValues(new Function<Iterable<Tuple2<Tuple2<String, String>, Integer>>, List<String>>() {
            @Override
            public List<String> call(Iterable<Tuple2<Tuple2<String, String>, Integer>> v1) throws Exception {
                // v1 ((辽宁,辽宁朝阳市果菜批发市场),37) 可遍历
                List<Tuple2<Tuple2<String, String>, Integer>> list = IterableUtils.toList(v1);
                list.sort(new Comparator<Tuple2<Tuple2<String, String>, Integer>>() {
                    @Override
                    public int compare(Tuple2<Tuple2<String, String>, Integer> o1, Tuple2<Tuple2<String, String>, Integer> o2) {
                        return o2._2 - o1._2;
                    }
                });
                List<String> strings = new ArrayList<>();
                for (Tuple2<Tuple2<String, String>, Integer> tuple2 : list) {
                    strings.add(tuple2._1._2);
                }
                if (strings.size() < 3) {
                    return strings;
                } else
                    return strings.subList(0, 3);
            }
        });
        System.out.println(rdd7.collect());
    }
}
